package hbase.operation;

import hbase.tools.Constants;
import hbase.tools.HbaseUtils;
import hbase.tools.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;

public class TableOperation {

    private static Connection conn = null;
    private static Admin admin = null;

    static{
        try {
            Configuration conf = HBaseConfiguration.create();

            conf.set(Constants.ZK_QUORUM,Constants.ZK_QUORUM_NODES);
            conf.set(Constants.ZK_CLIENT_PORT,Constants.ZK_PORT);

            conn =  ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建一个普通表，列族中的所有属性都使用默认值
     * @param tableName
     * @param columnFamilies
     */
    public static void createTable(String tableName,String... columnFamilies){
        createTable(tableName,true,columnFamilies);
    }

    /**
     * 创建一个普通表，列族中的所有属性都使用默认值
     * @param tableName
     * @param columnFamilies
     */
    public static void createTable(String tableName,boolean closeFlag,String... columnFamilies){
        List<Map<String, String>> cfs = new ArrayList<>();
        for (String columnFamily : columnFamilies) {
            Map<String,String> cf = new HashMap<>();
            cf.put("NAME",columnFamily);
            cfs.add(cf);
        }
        createTable(tableName,closeFlag,cfs);
    }

    /**
     * 创建一个普通表
     * @param tableName
     * @param columnFamilies
     */
    public static void createTable(String tableName, List<Map<String, String>> columnFamilies){
        createTable(tableName,true,columnFamilies);
    }

    /**
     * 创建一个普通表
     * @param tableName
     * @param columnFamilies
     */
    public static void createTable(String tableName,boolean closeFlag,List<Map<String, String>> columnFamilies){
        createTable(tableName,closeFlag,columnFamilies,null,null,null,0);
    }

    /**
     * 创建一个给定分区临界点的预分区表
     * @param tableName
     * @param columnFamilies
     * @param splitKeys 预分区的临界点
     */
    public static void createTable(String tableName, List<Map<String, String>> columnFamilies,String[] splitKeys){
        createTable(tableName,true,columnFamilies,splitKeys);
    }

    /**
     * 创建一个给定分区临界点的预分区表
     * @param tableName
     * @param columnFamilies
     * @param splitKeys 预分区的临界点
     */
    public static void createTable(String tableName,boolean closeFlag, List<Map<String, String>> columnFamilies,String[] splitKeys){
        createTable(tableName,closeFlag,columnFamilies,splitKeys,null,null,0);
    }

    /**
     * 创建一个给定起始和终止key以及分区数量的预分区表
     * @param tableName
     * @param columnFamilies
     * @param startKey  此key之前的为一个分区
     * @param endKey 此key之后的为一个分区
     * @param numRegions 分区的数量 要求必须大于等于3
     */
    public static void createTable(
            String tableName,
            List<Map<String, String>> columnFamilies,
            String startKey,
            String endKey,
            int numRegions
    ){
        createTable(tableName,true,columnFamilies,startKey,endKey,numRegions);
    }

    /**
     * 创建一个给定起始和终止key以及分区数量的预分区表
     * @param tableName
     * @param columnFamilies
     * @param startKey  此key之前的为一个分区
     * @param endKey 此key之后的为一个分区
     * @param numRegions 分区的数量 要求必须大于等于3
     */
    public static void createTable(
            String tableName,
            boolean closeFlag,
            List<Map<String, String>> columnFamilies,
            String startKey,
            String endKey,
            int numRegions
    ){
        createTable(tableName,closeFlag,columnFamilies,null,startKey,endKey,numRegions);
    }

    /**
     * 提供给当前类使用的创建表的方法
     * @param tableName
     * @param columnFamilies
     * @param splitKeys
     * @param startKey
     * @param endKey
     * @param numRegions
     */
    private static void createTable(
            String tableName,
            List<Map<String, String>> columnFamilies,
            String[] splitKeys,
            String startKey,
            String endKey,
            int numRegions
    ){
        createTable(tableName,true,columnFamilies,splitKeys,startKey,endKey,numRegions);
    }

    /**
     * 提供给当前类使用的创建表的方法
     * @param tableName
     * @param columnFamilies
     * @param splitKeys
     * @param startKey
     * @param endKey
     * @param numRegions
     */
    private static void createTable(
            String tableName,
            boolean closeFlag,
            List<Map<String, String>> columnFamilies,
            String[] splitKeys,
            String startKey,
            String endKey,
            int numRegions
    ){
        try {
            //如果表不存在可以创建表
            if(HbaseUtils.isTableNotExists(admin,tableName)){

                //创建一个表描述对象 TableDescriptorBuilder.ModifyableTableDescriptor
                TableDescriptorBuilder.ModifyableTableDescriptor td =
                        HbaseUtils.createTableDescriptor(tableName);

                //确保参数已经传递了列族信息
                if(columnFamilies!=null && columnFamilies.size()>0){

                    //遍历所有列族
                    for (Map<String, String> columnFamily : columnFamilies) {
                        //获取每个列族的描述对象
                        ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor cfd =
                                HbaseUtils.createCFDescriptor(columnFamily.get("NAME"));

                        //@TODO 这里需要判断列族的每个属性，这里只以VERSIONS为例，其他暂不考虑
                        if(columnFamily.containsKey("VERSIONS")){
                            cfd.setVersions(1,Integer.parseInt(columnFamily.get("VERSIONS")));
                        }

                        td.setColumnFamily(cfd);
                    }
                }else{
                    System.out.println("没有列族，无法创建表");
                    return;
                }

                //如果临界点不为空，创建有临界点的预分区表
                if(splitKeys!=null && splitKeys.length>0){
                    //创建一个临界点的字节数组类型的对象
                    byte[][] keys = new byte[splitKeys.length][];

                    //封装这个临界点的二维数组，第一维表示索引位置，第二位表示的是临界点的字节数组类型
                    for (int i = 0; i < keys.length; i++) {
                        keys[i] = Bytes.toBytes(splitKeys[i]);
                    }

                    admin.createTable(td,keys);
                    return;
                }

                //如果起始和终止的key都不为空且分区数量大于0
                if(StringUtils.isStringNotEmpty(startKey) && StringUtils.isStringNotEmpty(endKey) && numRegions>2){
                    admin.createTable(td,Bytes.toBytes(startKey),Bytes.toBytes(endKey),numRegions);
                    return;
                }

                //如果不是预分区的情况，直接走默认创建表
                admin.createTable(td);
            }else {
                //如果表存在，给提示
                System.out.println(tableName+" 表已经存在");
            }
        } catch (IOException e) {
                throw new RuntimeException(e);
        } finally {
            if(closeFlag){
                HbaseUtils.closed(conn,admin);
            }
        }
    }

    /**
     * 修改表的列族信息
     * @param tableName
     * @param columnFamilies 如果是List类型，就是删除列族；如果是Map类型，就是添加或修改列族
     */
    public static void alterTable(String tableName,Object... columnFamilies){
        try {
            //如果存在表，可以做修改
            if(HbaseUtils.isTableExists(admin,tableName)){

                //获取当前表的描述对象 TableDescriptorBuilder.ModifyableTableDescriptor
                TableDescriptorBuilder.ModifyableTableDescriptor td = HbaseUtils.getTableDescriptor(admin,tableName);

                //遍历所有的列族
                for (Object columnFamily : columnFamilies) {
                    //如果 columnFamilies 是 Map，表示添加或修改属性
                    if(columnFamily instanceof Map){
                        Map<String,String> cf = (Map<String, String>) columnFamily;
                        String cfName = cf.get("NAME");

                        if(td.hasColumnFamily(Bytes.toBytes(cfName))){
                            //获取当前需要修改的列族的描述对象
                            ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor cfd =
                                    (ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor)td.getColumnFamily(Bytes.toBytes(cfName));
                            //@TODO 这里需要判断列族的每个属性，这里只以VERSIONS为例，其他暂不考虑
                            if(cf.containsKey("VERSIONS")){
                                cfd.setVersions(1,Integer.parseInt(cf.get("VERSIONS")));
                            }
                            if(cf.containsKey("BLOOMFILTER")){
                                String bloom = cf.get("BLOOMFILTER");
                                if(bloom.equals("ROWCOL")){
                                    cfd.setBloomFilterType(BloomType.ROWCOL);
                                }else if(bloom.equals("ROW")){
                                    cfd.setBloomFilterType(BloomType.ROW);
                                }else if(bloom.equals("NONE")){
                                    cfd.setBloomFilterType(BloomType.NONE);
                                }
                            }
                            admin.modifyColumnFamily(TableName.valueOf(tableName),cfd);
                        }else{
                            //创建当前需要添加的列族的描述对象
                            ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor cfd =
                                    HbaseUtils.createCFDescriptor(cfName);
                            //@TODO 这里需要判断列族的每个属性，这里只以VERSIONS为例，其他暂不考虑
                            if(cf.containsKey("VERSIONS")){
                                cfd.setVersions(1,Integer.parseInt(cf.get("VERSIONS")));
                            }
                            admin.addColumnFamily(TableName.valueOf(tableName),cfd);
                        }
                    }else if(columnFamily instanceof String){
                        //如果 columnFamilies 是 String，表示删除属性
                        String cfName = (String) columnFamily;
                        if(td.hasColumnFamily(Bytes.toBytes(cfName))){
                            admin.deleteColumnFamily(TableName.valueOf(tableName),Bytes.toBytes(cfName));
                        }else{
                            System.out.println(cfName + " 列族不存在，无法删除");
                        }
                    }
                }
            }else{
                System.out.println(tableName+" 表不存在");
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            HbaseUtils.closed(conn,admin);
        }
    }

    /**
     * 修改表的可用和禁用状态
     * @param status ture为改成可用；false为改成禁用
     * @param tableNames
     */
    public static void modifyTableStatus(boolean status,String... tableNames){
        modifyTableStatus(status,true,tableNames);
    }

    /**
     * 修改表的可用和禁用状态
     * @param status ture为改成可用；false为改成禁用
     * @param closeFlag true表示关闭资源；false表示不关闭资源
     * @param tableNames
     */
    public static void modifyTableStatus(boolean status,boolean closeFlag,String... tableNames){
        try {
            for (String tableName : tableNames) {
                if(HbaseUtils.isTableExists(admin,tableName)){
                    TableName tbName = TableName.valueOf(tableName);
                    //如果当前表是禁用状态且要变成可用状态时
                    if(status && admin.isTableDisabled(tbName)){
                        admin.enableTable(tbName);
                    }else if(!status && admin.isTableEnabled(tbName)){
                        //如果当前表是可用状态且要变成禁用状态时
                        admin.disableTable(tbName);
                    }else{
                        System.out.println(tableName+" 表的状态无需改变");
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(closeFlag){
                HbaseUtils.closed(conn,admin);
            }
        }
    }

    /**
     * 给正则表达式，对所有符合条件的表修改状态
     * @param status
     * @param regex
     */
    public static void modifyTableStatusByPattern(boolean status,String regex){
        modifyTableStatusByPattern(status,true,regex);
    }

    /**
     * 给正则表达式，对所有符合条件的表修改状态
     * @param status
     * @param closeFlag
     * @param regex
     */
    public static void modifyTableStatusByPattern(boolean status,boolean closeFlag,String regex){
        try {
            //通过正则表达式查找所有符合条件的表
            List<TableDescriptor> tdList = admin.listTableDescriptors(Pattern.compile(regex));

            //遍历所有的表，对其进行修改状态
            for (TableDescriptor td : tdList) {
                TableName tableName = td.getTableName();
                //如果当前表是禁用状态且要变成可用状态时
                if(status && admin.isTableDisabled(tableName)){
                    admin.enableTable(tableName);
                }else if(!status && admin.isTableEnabled(tableName)){
                    //如果当前表是可用状态且要变成禁用状态时
                    admin.disableTable(tableName);
                }else{
                    System.out.println(tableName+" 表的状态无需改变");
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if(closeFlag){
                HbaseUtils.closed(conn,admin);
            }
        }
    }


    /**
     * 获取表信息
     * @param tableNames 如果不传参，查询所有表；如果传参，按指定表查询
     */
    public static void getTableInfo(String... tableNames){
        try {
            //查询指定表信息
            if(tableNames!=null && tableNames.length>0){
                for (String tableName : tableNames) {
                    //如果当前表存在，获取表信息，进行输出
                    if(HbaseUtils.isTableExists(admin,tableName)){
                        TableDescriptor descriptor = admin.getDescriptor(TableName.valueOf(tableName));
                        System.out.print(descriptor.getTableName()+" 表有以下列族：");
                        Set<byte[]> cfNames = descriptor.getColumnFamilyNames();
                        for (byte[] cfName : cfNames) {
                            System.out.print(new String(cfName)+" ");
                        }
                        System.out.println("\n=======================================");
                    }else{
                        System.out.println(tableName+" 表不存在");
                    }
                }
            }else{
                //查询所有表信息
                List<TableDescriptor> tds = admin.listTableDescriptors();
                for (TableDescriptor td : tds) {
                    System.out.print(td.getTableName()+" 表有以下列族：");
                    Set<byte[]> cfNames = td.getColumnFamilyNames();
                    for (byte[] cfName : cfNames) {
                        System.out.print(new String(cfName)+" ");
                    }
                    System.out.println("\n=======================================");
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            HbaseUtils.closed(conn,admin);
        }
    }

    /**
     * 删除所有表
     * @param isDeleteAll
     */
    public static void dropTable(boolean isDeleteAll){
        dropTable(true,null);
    }

    /**
     * 删除指定表
     * @param tableNames
     */
    public static void dropTable(String... tableNames){
        dropTable(false,tableNames);
    }

    /**
     * 封装当前类可用的删除表方法
     * @param tableNames
     */
    private static void dropTable(boolean isDeleteAll,String... tableNames){
        try {
            //如果 isDeleteAll 为 true，表示删除所有表；如果为 false，表示删除指定表
            if(isDeleteAll){
                List<TableDescriptor> tds = admin.listTableDescriptors();
                for (TableDescriptor td : tds) {
                    admin.disableTable(td.getTableName());
                    admin.deleteTable(td.getTableName());
                }
            }else{
                if(tableNames!=null && tableNames.length>0){
                    for (String tableName : tableNames) {
                        if(HbaseUtils.isTableExists(admin,tableName)){
                            TableName tbName = TableName.valueOf(tableName);
                            admin.disableTable(tbName);
                            admin.deleteTable(tbName);
                        }else{
                            System.out.println(tableName+" 表不存在");
                        }
                    }
                }else{
                    System.out.println("没有指定要删除的表");
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        List<Map<String,String>> cfList = new ArrayList<>();
        //第一个列族
        Map<String,String> cfMap1 = new HashMap<>();
        cfMap1.put("NAME","baseInfo");
//        cfMap1.put("VERSIONS","10");
        cfMap1.put("BLOOMFILTER","ROWCOL");
        //第二个列族
        Map<String,String> cfMap2 = new HashMap<>();
        cfMap2.put("NAME","mdddddInfo");
        //把两个列族添加到列族列表中
        cfList.add(cfMap1);
        cfList.add(cfMap2);


//        TableOperation.createTable("s1",cfList);
//        TableOperation.createTable("s2",cfList,new String[]{"10","20","30"});
//        TableOperation.createTable("s3",cfList,"100","500",5);
//        TableOperation.createTable("s4",cfList,"100","500",2);
//        TableOperation.createTable("s5",cfList);
//        TableOperation.alterTable("s5",cfMap1,cfMap2);
//        TableOperation.alterTable("s5","info","mdddddInfo","moreInfo");
//        TableOperation.modifyTableStatus(true,"s1","s2","s3","s5");
//        TableOperation.getTableInfo("s2","s3","s4");
//        TableOperation.dropTable("s1","s3");
        TableOperation.dropTable(true);


    }

}
